package com.shujia.flink.tf;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple22;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo5Reduce {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> linesDS = env.socketTextStream("master", 8888);

        //展开
        DataStream<Tuple2<String, Integer>> kvDS = linesDS
                .flatMap((line, out) -> {
                    for (String word : line.split(",")) {
                        //转换成kv发送到下游
                        out.collect(Tuple2.of(word, 1));
                    }
                }, Types.TUPLE(Types.STRING, Types.INT));

        //分组
        KeyedStream<Tuple2<String, Integer>, String> keyByDS = kvDS.keyBy(kv -> kv.f0);

        DataStream<Tuple2<String, Integer>> countDS = keyByDS
                .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> kv1,
                                                          Tuple2<String, Integer> kv2) throws Exception {
                        //kv1和kv2的key是一样的
                        String word = kv1.f0;
                        int count = kv1.f1 + kv2.f1;
                        return Tuple2.of(word, count);
                    }
                });

        ///countDS.print();


        //kv1是上一次的计算结果（状态）
        //jv2是当前需要计算的数据
        keyByDS.reduce((kv1, kv2) -> {
            //System.out.println(kv1);
            //System.out.println(kv2);
            //kv1和kv2的key是一样的
            String word = kv1.f0;
            int count = kv1.f1 + kv2.f1;
            return Tuple2.of(word, count);
        });//.print();

        //求和
        keyByDS.sum(1).print();

        env.execute();

    }
}
